package com.inspinia.base.mq;

import com.google.common.collect.Maps;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * 生产者
 * Created by veblen on 16/5/3.
 */
public class KafkaProducerHelper {

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerHelper.class);

    /**
     * 配置文件
     */
    private static Properties props;
    /**
     * template的map,key是topic,valude是KafkaTemplate
     */
    private static Map<String, KafkaTemplate<String, String>> templateMap;

    private static void loadProperties() {

        //加载配置文件
        props = new Properties();
        props.put("bootstrap.servers", KafkaConfig.BOOTSTRAP_SERVERS);
        props.put("acks", KafkaConfig.ACKS);
        props.put("retries", KafkaConfig.RETRIES);
        props.put("batch.size",KafkaConfig.BATCH_SIZE);
        props.put("linger.ms", KafkaConfig.LINGER_MS);
        props.put("buffer.memory",KafkaConfig.BUFFER_MEMORY);
        props.put("key.serializer", KafkaConfig.KEY_DESERIALIZER);
        props.put("value.serializer",KafkaConfig.VALUE_DESERIALIZER);
    }


    /**
     * 创建kafkaTemplate
     *
     * @return
     */
    private static KafkaTemplate<String, String> createTemplate() {
        if (null == props) {
            loadProperties();
        }

        Map map = Maps.fromProperties(props);
        ProducerFactory<String, String> pf =
                new DefaultKafkaProducerFactory(map);
        return new KafkaTemplate<>(pf);
    }


    /**
     * 发送数据
     *
     * @param topic 类目
     * @param data  需要发送的数据
     */
    public static void send(String topic, String data) {
        if (templateMap == null) {
            templateMap = new HashMap<>();
        }
        KafkaTemplate<String, String> kafkaTemplate = templateMap.get(topic);
        if (kafkaTemplate == null) {
            kafkaTemplate = createTemplate();
            templateMap.put(topic, kafkaTemplate);
        }
        kafkaTemplate.setDefaultTopic(topic);
        kafkaTemplate.send(topic,data);
        kafkaTemplate.flush();
    }


}
